package com.hadoop.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

import java.io.IOException;

/**
 * <p></p>
 *
 * @author Andy
 * @date 2017/10/21
 */
public class TotalSortMR {

    public static int runTotalSortJob(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Path inputPath = new Path(args[0]);
        Path outputPath = new Path(args[1]);
        Path partitionFile = new Path(args[2]);
        int reduceNum = Integer.valueOf(args[3]);

        //
        InputSampler.RandomSampler<Text, Text> sampler = new InputSampler.RandomSampler(1, 3000, 10);

        Configuration conf = new Configuration();
        TotalOrderPartitioner.setPartitionFile(conf, partitionFile);

        Job job = Job.getInstance(conf);
        job.setJobName("Total-sort");
        job.setJarByClass(TotalSortMR.class);
        job.setInputFormatClass(KeyValueTextInputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setNumReduceTasks(reduceNum);

        //
        job.setPartitionerClass(TotalOrderPartitioner.class);

        FileInputFormat.addInputPath(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(conf).delete(outputPath, true);

        //
        InputSampler.writePartitionFile(job, sampler);

        return job.waitForCompletion(true) ? 0 : 1;
//        return 0;
    }

    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
        System.exit(runTotalSortJob(args));
    }

}
